package dongshi.daddy.zab.leadercampaign;

import cn.hutool.json.JSONUtil;
import dongshi.daddy.zab.leadercampaign.bo.Vote;

import java.io.*;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class LeaderCampaignTest {
    private static CountDownLatch mainNotExits = new CountDownLatch(1);
    private static CountDownLatch leaderDownCdl = new CountDownLatch(1);
    private static CountDownLatch selfVoteReceivedCdl = new CountDownLatch(2);
    private static CountDownLatch receiveBChangeVoteCdl = new CountDownLatch(2);
    private static String NODE_A_NAME = "nodeA";
    private static int NODE_A_PORT = 6668;
    private static String NODE_B_NAME = "nodeB";
    private static int NODE_B_PORT = 6669;
    private static String NODE_C_NAME = "nodeC";
    private static int NODE_C_PORT = 6670;

    public static void main(String[] args) throws Exception {
        Thread nodeA = new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + " 准备好了！");
                // 模拟leader宕机
                Thread.sleep(1000);
                leaderDownCdl.countDown();
                System.out.println("");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, NODE_A_NAME);
        nodeA.start();

        Thread nodeB = new Thread(() -> {
            try {
                String whoAmI = NODE_B_NAME;
                int myid = 2;
                int epoch = 1;
                int lastZxId = 101;

                ServerSocket serverSocket = new ServerSocket(NODE_B_PORT);
                // 所有有效的投票
                List<Vote> validateVoteList = new ArrayList<>();
                new Thread(() -> {
                    while (true) {
//                        counter++;
                        try {
                            String msg = receiveMsg(serverSocket);
                            Vote vote = JSONUtil.toBean(msg, Vote.class);
                            if (validateVoteList.size() == 0) { // 收到的是自己的投票
                                System.out.println(whoAmI + "收到自己的选票: " + msg);
                                validateVoteList.add(vote);
                                selfVoteReceivedCdl.countDown();
                            } else if (validateVoteList.size() == 1) { // 收到来自对方的消息
//                                System.out.println(whoAmI + "收到选票: " + msg);
                                validateVoteList.add(vote);
                                // leader PK
                                Vote oldLeaderVote = validateVoteList.get(0);
                                if (oldLeaderVote.getEpoch() < vote.getEpoch()
                                        || oldLeaderVote.getLastZxId() < vote.getLastZxId()
                                        || oldLeaderVote.getMyid() < vote.getMyid()) {
//                                    validateVoteList.add(vote);
                                    System.out.println(whoAmI + " 需要变更领导者选票信息，变更前为：" + JSONUtil.toJsonStr(oldLeaderVote) + ", 变更后为：" + JSONUtil.toJsonStr(vote));
                                    // 发送新的选票信息
                                    try {
                                        sendMsg(JSONUtil.toJsonStr(vote), NODE_B_PORT);
                                        sendMsg(JSONUtil.toJsonStr(vote), NODE_C_PORT);
                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                }
                            } else if (validateVoteList.size() == 2) { // 节点B变更leader选票消息
                                validateVoteList.add(vote);
                                System.out.println(whoAmI + "收到节点B变更leader选票信息：" + JSONUtil.toJsonStr(vote));
                                receiveBChangeVoteCdl.countDown();
                            }

                            // 按照如下规则排序
                            /*
                            优先检查任期编号（Epoch），任期编号大的节点作为领导者；
                            如果任期编号相同，比较事务标识符的最大值，值大的节点作为领导者；
                            如果事务标识符的最大值相同，比较集群 ID，集群 ID 大的节点作为领导者。
                             */
                            Collections.sort(validateVoteList, (o1, o2) -> {
                                if (o1.getEpoch() != o2.getEpoch()) return o2.getEpoch() - o1.getEpoch();
                                if (o1.getLastZxId() != o2.getLastZxId()) return o2.getLastZxId() - o1.getLastZxId();
                                if (o1.getMyid() != o2.getMyid()) return o2.getMyid() - o1.getMyid();
                                return 0;
                            });
//                            System.out.println(whoAmI + "当前有效选票信息是: " + validateVoteList);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");

                leaderDownCdl.await();
                // 创建投票信息
                Vote vote = new Vote();
                vote.setEpoch(epoch);
                vote.setLastZxId(lastZxId);
                vote.setMyid(myid);
                vote.setWhoAmI(whoAmI);
                String voteStr = JSONUtil.toJsonStr(vote);
                System.out.println(Thread.currentThread().getName() + " 检测到leader故障，进入following状态！,并生成投票信息：" + voteStr);
                // 先发送选票信息给自己
                sendMsg(voteStr, NODE_B_PORT);

                // 发消息给对方
                selfVoteReceivedCdl.await();
                sendMsg(voteStr, NODE_C_PORT);;

                // 选出新leader
                receiveBChangeVoteCdl.await();
                System.out.println(whoAmI + " 没有收到大多数选票，成为follower，转换为FOLLOWING状态");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, NODE_B_NAME);
        nodeB.start();

        Thread nodeC = new Thread(() -> {
            try {
                int myid = 3;
                int epoch = 1;
                int lastZxId = 102;
                String whoAmI = NODE_C_NAME;
                ServerSocket serverSocket = new ServerSocket(NODE_C_PORT);
                new Thread(() -> {
                    // 所有有效的投票
                    List<Vote> validateVoteList = new ArrayList<>();
                    while (true) {
//                        counter++;
                        try {
                            String msg = receiveMsg(serverSocket);
                            Vote vote = JSONUtil.toBean(msg, Vote.class);
                            if (validateVoteList.size() == 0) { // 收到的是自己的投票
                                System.out.println(whoAmI + "收到自己的选票: " + msg);
                                validateVoteList.add(vote);
                                selfVoteReceivedCdl.countDown();
                            } else if (validateVoteList.size() == 1) { // 收到来自对方的消息
//                                System.out.println(whoAmI + "收到选票: " + msg);
                                validateVoteList.add(vote);
                                // leader PK
                                Vote oldLeaderVote = validateVoteList.get(0);
                                if (oldLeaderVote.getEpoch() < vote.getEpoch()
                                        || oldLeaderVote.getLastZxId() < vote.getLastZxId()
                                        || oldLeaderVote.getMyid() < vote.getMyid()) {
//                                    validateVoteList.add(vote);
                                    System.out.println(whoAmI + " 需要变更领导者选票信息，变更前为：" + JSONUtil.toJsonStr(oldLeaderVote) + ", 变更后为：" + JSONUtil.toJsonStr(vote));
                                    // 发送新的选票信息
                                    try {
                                        sendMsg(JSONUtil.toJsonStr(vote), NODE_B_PORT);
                                        sendMsg(JSONUtil.toJsonStr(vote), NODE_C_PORT);
                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                }
                            } else if (validateVoteList.size() == 2) { // 节点B变更leader选票消息
                                validateVoteList.add(vote);
                                System.out.println(whoAmI + "收到节点B变更leader选票信息：" + JSONUtil.toJsonStr(vote));
                                receiveBChangeVoteCdl.countDown();
                            }

                            // 按照如下规则排序
                            /*
                            优先检查任期编号（Epoch），任期编号大的节点作为领导者；
                            如果任期编号相同，比较事务标识符的最大值，值大的节点作为领导者；
                            如果事务标识符的最大值相同，比较集群 ID，集群 ID 大的节点作为领导者。
                             */
                            Collections.sort(validateVoteList, (o1, o2) -> {
                                if (o1.getEpoch() != o2.getEpoch()) return o1.getEpoch() - o2.getEpoch();
                                if (o1.getLastZxId() != o2.getLastZxId()) return o1.getLastZxId() - o2.getLastZxId();
                                if (o1.getMyid() != o2.getMyid()) return o1.getMyid() - o2.getMyid();
                                return 0;
                            });
//                            System.out.println(whoAmI + "当前有效选票信息是: " + validateVoteList);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");

                leaderDownCdl.await();
                // 创建投票信息
                Vote vote = new Vote();
                vote.setEpoch(epoch);
                vote.setLastZxId(lastZxId);
                vote.setMyid(myid);
                vote.setWhoAmI(whoAmI);
                String voteStr = JSONUtil.toJsonStr(vote);
                System.out.println(Thread.currentThread().getName() + " 检测到leader故障，进入following状态！,并生成投票信息：" + voteStr);
                // 先发送选票信息给自己
                sendMsg(voteStr, NODE_C_PORT);;

                // 发消息给对方
                selfVoteReceivedCdl.await();
                sendMsg(voteStr, NODE_B_PORT);;

                // 选出新leader
                receiveBChangeVoteCdl.await();
                System.err.println(whoAmI + " 收到大多数选票，成为leader，转换为LEADING状态");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, NODE_C_NAME);
        nodeC.start();


        mainNotExits.await();
    }

    public static void sendMsg(String sendMsg, int port) throws Exception {
        Socket socket = new Socket();
        // 设置超时不生效，这里手动休眠
        Thread.sleep(200);
        socket.connect(new InetSocketAddress("127.0.0.1", port));
        OutputStream os = socket.getOutputStream();
        PrintWriter pw = new PrintWriter(os);
        pw.println(sendMsg);
        pw.flush();
        socket.close();
    }

    public static String receiveMsg(ServerSocket serverA) throws IOException {
        Socket socketA = serverA.accept();
        InputStream is = socketA.getInputStream();
        InputStreamReader isr = new InputStreamReader(is);
        BufferedReader br = new BufferedReader(isr);
        while (true) {
            try {
                String s = br.readLine();
                if (s == null) {
                    break;
                    // 模拟第二轮发送消息，这里发送的是第一轮收到的消息
                }
                return s;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return "";
    }

}
